package org.example.netty.zk.lock;

import org.apache.commons.lang3.StringUtils;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.example.zk.ZkClient;

import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * zk锁的对象
 * 分布式锁的类
 * 一个线程代表一个ZkLock类
 */
public class ZkLockImpl implements ZkLock{

    private final String LOCK_NAME = "/test/lock";

    private final String LOCK_SUBNAME_PREFIX = LOCK_NAME + "/node-";

    private Thread thread;

    private AtomicInteger count = new AtomicInteger(0);

    private String nodeId;

    private String shortNodeId;

    public ZkLockImpl() {
        // TODO 避免多实例同时操作
        synchronized (ZkLockImpl.class) {
            if (!ZkClient.instance.nodeExists(LOCK_NAME)) {
                ZkClient.instance.createNode(LOCK_NAME);
            }
        }
    }

    @Override
    public void lock() {
        // TODO 重入锁处理
        synchronized (this) {
            if (count.get() == 0) {
                thread = Thread.currentThread();
                count.getAndIncrement();
            }
            else {
                if (thread != Thread.currentThread()) {
                    return;
                }
                count.getAndIncrement();
                return;
            }
        }

        // TODO 尝试加锁
        boolean locked = false;

        // TODO 创建临时节点进行枷锁
        nodeId = ZkClient.instance.createSeqNode(LOCK_SUBNAME_PREFIX);
        shortNodeId = "node-" + getNumberStr(nodeId);

        // TODO 先获取子节点，判断是否获取锁成功
        List<String> childNodes = ZkClient.instance.getChildrenNode(LOCK_NAME);

        if (checkNodes(childNodes))
            return;

        while (!locked) {
            await();

            // TODO 继续获取子节点，然后进行最小化处理
            childNodes = ZkClient.instance.getChildrenNode(LOCK_NAME);

            locked = checkNodes(childNodes);
        }
    }

    /**
     * 判断是否子节点列表第一个
     * @param childNodes
     * @return
     */
    public boolean checkNodes(List<String> childNodes) {
        Collections.sort(childNodes);
        // TODO 处理掉nodeId的前缀
        if (shortNodeId.equals(childNodes.get(0)))
            return true;
        return false;
    }

    /**
     * 线程等待，等待前一个节点被删除
     */
    public void await() {
        // TODO
        CountDownLatch latch = new CountDownLatch(1);

        List<String> childNodes = ZkClient.instance.getChildrenNode(LOCK_NAME);
        // TODO 对childNodes进行排序
        Collections.sort(childNodes);
        int curIndex = childNodes.indexOf(shortNodeId);
        // TODO 等你被删除，但是还没有执行到getData就已经删除了
        String prePath = LOCK_NAME + "/" + childNodes.get(curIndex - 1);

        try {
            ZkClient.instance.client.getData().usingWatcher(new Watcher() {
                @Override
                public void process(WatchedEvent watchedEvent) {
                    // TODO 前一个节点删除事件
                    if (watchedEvent.getType() == Event.EventType.NodeDeleted) {
                        latch.countDown();
                    }
                }
            }).forPath(prePath);
        }
        catch (KeeperException.NoNodeException e) {
            // TODO 节点已经被删除
            System.out.println(prePath + "已经被删除");
             latch.countDown();
        }
        catch (Exception e) {
            e.printStackTrace();
        }

        try {
            // TODO 等1秒中
            System.out.println("等待" + prePath + "被删除");
            latch.await();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 释放锁
     */
    public void releaseLock() {
        // TODO 只有本线程才能释放
        if (thread != Thread.currentThread())
            return;

        int tmp = count.decrementAndGet();

        if (tmp > 0)
            return;

        ZkClient.instance.deleteNode(nodeId);
    }

    /**
     * 获取node的编号
     *
     * @param path
     * @return
     */
    public long getNumber(String path) {
        int preIndex1 = path.lastIndexOf(LOCK_SUBNAME_PREFIX);
        long seqNo1 = Long.parseLong(path.substring(preIndex1 + LOCK_SUBNAME_PREFIX.length()));
        return seqNo1;
    }

    /**
     * 获取字符串类的编号
     * @param path
     * @return
     */
    public String getNumberStr(String path) {
        int preIndex1 = path.lastIndexOf(LOCK_SUBNAME_PREFIX);
        return path.substring(preIndex1 + LOCK_SUBNAME_PREFIX.length());
    }
}